package cn.com.infcn.util.mss;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

import org.ansj.app.keyword.Keyword;
import org.ansj.app.summary.SummaryComputer;
import org.ansj.app.summary.TagContent;
import org.ansj.app.summary.pojo.Summary;
import org.ansj.splitWord.analysis.DicAnalysis;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.get.GetRequestBuilder;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MoreLikeThisQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.QueryStringQueryBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Order;
import org.elasticsearch.search.aggregations.bucket.terms.TermsBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.nlpcn.commons.lang.util.StringUtil;
import org.nutz.mvc.annotation.Param;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;

//import cn.com.infcn.util.query.parse.ParseFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;

import cn.com.infcn.po.Restful;

/**
 * 搜索api
 *
 * @author Ansj
 */
@Repository
public class MssSearchApi {

	@Autowired
	private TransportClient mssEsClient;

	/**
	 * @param tableNames
	 *            tableNames 多个表名按照,号分割table1,table2
	 * @param size
	 *            返回结果数
	 * @param query
	 *            查询语句title:aa
	 * @param filter
	 *            过滤语句不参与打分
	 * @param field
	 *            过滤语句不参与打分
	 *            nested|FilesCat_f|FilesCat_f_1|FilesCat_f_2|FilesCat_f_3,term|aaa|count,date_histogram|field|1d|yyyyMMdd
	 * @return 返回结果，如果查询条件在20个小时内重复返回缓存内容
	 * @throws IOException
	 * @throws InterruptedException
	 * @throws ExecutionException
	 */
	public Restful aggregationsByQuery(String tableNames, int featchSize, int size, String query, String filter,
			String[] field) throws IOException, InterruptedException, ExecutionException {
		SearchResponse searchResponse = _aggregations(tableNames, featchSize, size, field, query, filter);
		return Restful.instance(JSON.parse(searchResponse.toString()));
	}

	/**
	 * @param tableNames
	 *            多个表名按照,号分割table1,table2
	 * @param from
	 *            起始位置
	 * @param size
	 *            返回结果数
	 * @param defaultOperator
	 *            默认的运算符(默认or): and|or
	 * @param query
	 *            查询语句title:aa
	 * @param filter
	 *            过滤语句不参与打分
	 * @param heighlight
	 *            高亮
	 *            [{keyword:['体育'],query:'体育',begin:'<begin>',end:'<end>',fields:[{field:'content',dist_field:'content_hl',size:0}]},{keyword:['体育'],query:'体育',begin:'<begin>',end:'<end>',fields:[{field:'content',dist_field:'content_hl',size:0}]}]
	 * @param includes
	 *            返回结果包含哪些
	 * @param excludes
	 *            返回结果不包含哪些
	 * @param distinct
	 *            去重
	 *            {"fetchCount":100,"fields":[{"field":"language","type":"hash"}],"mergeList":true}
	 * @param sort
	 *            排序 sort pubyear|asc,_score|desc
	 * @param explain
	 *            是否调试状态
	 * @return
	 */
	public Restful searchByQuery(String tableNames, int from, int size, String defaultOperator, String query,
			String filter, String heighlight, String includes, String excludes, String distinct, String sort,
			boolean explain) {
		Object search = _search(tableNames, from, size, defaultOperator, explain, includes, excludes, distinct, query,
				filter, sort);
		JSONObject json = heighligth(JSON.parseObject(search.toString()), heighlight);
		return Restful.instance(json);
	}

	/**
	 * 相关推荐
	 *
	 * @param tableNames
	 *            表名
	 * @param doc
	 *            需要推荐的文档
	 * @param size
	 *            返回结果数, 默认10
	 * @param includese
	 *            返回结果包含哪些字段
	 * @return
	 * @throws IOException
	 */
	public Restful recommend(@Param("tableNames") String[] tableNames, @Param("doc") String doc,
			@Param(value = "size", df = "10") int size, @Param("includes") String[] includes) throws IOException {
		if (tableNames == null || tableNames.length < 1) {
			throw new IllegalArgumentException("arg[tableNames] is empty");
		}
		if (StringUtils.isBlank(doc)) {
			throw new IllegalArgumentException("arg[doc] is empty");
		}
		if (size < 0) {
			throw new IllegalArgumentException("invalid arg[size]: " + size);
		}

		MoreLikeThisQueryBuilder mlt = QueryBuilders.moreLikeThisQuery().minTermFreq(1).like(
				new MoreLikeThisQueryBuilder.Item().doc(XContentFactory.jsonBuilder().map(JSON.parseObject(doc))));
		SearchRequestBuilder search = mssEsClient.prepareSearch(tableNames)
				.setSearchType(SearchType.DFS_QUERY_THEN_FETCH).setSize(size).setFetchSource(includes, null)
				.setQuery(mlt);

		return Restful.instance(JSONObject.parse(search.get().toString()));
	}

	/**
	 * 查询所有表的记录数，如果库表没有被创建，则为0
	 *
	 * @param tableNames
	 *            split by ","
	 * @return all tables count
	 * @throws ExecutionException
	 * @throws InterruptedException
	 */

	public Restful count(String[] tableNames) throws InterruptedException, ExecutionException {

		long start = System.currentTimeMillis();

		List<String> indexes = new ArrayList<>();

		for (String tableName : tableNames) {
			if (StringUtil.isBlank(tableName)) {
				continue;
			}
			if (mssEsClient.admin().indices().prepareExists(tableName).get().isExists()) {
				indexes.add(tableName);
			}
		}

		JSONObject result = null;

		if (indexes.size() > 0) {
			SearchResponse searchResponse = mssEsClient
					.prepareSearch(indexes.toArray(new String[indexes.size()])).setSize(0).execute().get();
			result = JSONObject.parseObject(searchResponse.toString());
		} else {
			result = new JSONObject();
			JSONObject hits = new JSONObject();
			hits.put("total", 0);
			result.put("hits", hits);
			result.put("took", System.currentTimeMillis() - start);
		}

		return Restful.instance(result);
	}

	private JSONObject heighligth(JSONObject json, String heighligth) {

		if (StringUtil.isBlank(heighligth)) {
			return json;
		}

		Object parse = JSON.parse(heighligth);

		if (parse instanceof JSONObject) {
			heighligth(json, (JSONObject) parse);
		} else if (parse instanceof JSONArray) {
			for (Object hl : (JSONArray) parse) {
				heighligth(json, (JSONObject) hl);
			}
		} else {
			// log.error(heighligth + " parse err");
			System.out.println(heighligth + " parse err");
		}

		return json;
	}

	private void heighligth(JSONObject json, JSONObject hl) {
		JSONArray fields = hl.getJSONArray("fields");

		boolean single = hl.getBooleanValue("single");

		if (fields == null || fields.size() == 0) {
			return;
		}

		Set<String> kwSet = new HashSet<>();

		JSONArray kwJar = hl.getJSONArray("keyword");

		if (kwJar != null && kwJar.size() > 0) {
			kwJar.forEach(k -> kwSet.add((String) k));
		}

		String query = hl.getString("query");

		boolean ellipsis = hl.getBooleanValue("ellipsis");

		if (StringUtil.isNotBlank(query)) {
			DicAnalysis.parse(query).forEach(t -> {
				if (StringUtil.isNotBlank(t.getName())) {
					if (!",".equals(t.getName()) && (single || t.getName().length() > 0)) { // skip lenth 1
						kwSet.add(t.getName());
					}
				}
			});
		}

		if (kwSet == null || kwSet.size() == 0) {
			kwSet.add("ANSJ_NOT_HAVE_CONTENT");
		}

		List<Keyword> keywords = kwSet.stream().map(s -> new Keyword(s, Math.pow(10, s.length())))
				.collect(Collectors.toList());

		final int len = fields.size();

		String begin = hl.getString("begin");
		String end = hl.getString("end");

		TagContent tc = new TagContent(begin, end);
		json.getJSONObject("hits").getJSONArray("hits").forEach(j -> {
			JSONObject job = ((JSONObject) j).getJSONObject("_source");
			for (int i = 0; i < len; i++) {
				JSONObject hj = fields.getJSONObject(i);

				String field = hj.getString("field");
				String value = job.getString(field);
				if (StringUtil.isBlank(value)) {
					continue;
				}
				int size = hj.getIntValue("size");
				String distField = (String) hj.getOrDefault("dist_field", field);
				SummaryComputer sc = new SummaryComputer(size <= 0 ? Integer.MAX_VALUE : size, true, "", value);
				Summary summary = sc.toSummary(keywords);

				if (ellipsis) {
					StringBuilder newSummary = new StringBuilder();
					if (!value.startsWith(summary.getSummary())) {
						newSummary.append("…");
					}
					newSummary.append(summary.getSummary());
					if (!value.endsWith(summary.getSummary())) {
						newSummary.append("…");
					}
					summary = new Summary(summary.getKeyWords(), newSummary.toString());
				}

				String tagContent = tc.tagContent(summary);
				job.put(distField, tagContent);
			}
		});

	}

	private SearchResponse _aggregations(String tableNames, int featchSize, int size, String[] field, String queryDsl,
			String filterDsl) {

		// log.info("tables: {} ;field: {} ; query: {} ; filter: {}", tableNames,
		// Arrays.toString(field), queryDsl, filterDsl);

		BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

		if (StringUtil.isNotBlank(filterDsl)) {
			boolQuery.filter(QueryBuilders.queryStringQuery(filterDsl).lowercaseExpandedTerms(false));
		}

		if (StringUtil.isNotBlank(queryDsl)) {
			boolQuery.must(QueryBuilders.queryStringQuery(queryDsl).lowercaseExpandedTerms(false));
		}

		return _aggregations(tableNames, featchSize, size, field, boolQuery);
	}

	/**
	 * es query查询
	 *
	 * @param tableNames
	 * @param size
	 * @param field
	 * @param boolQuery
	 * @return
	 */
	private SearchResponse _aggregations(String tableNames, int featchSize, int size, String[] field,
			QueryBuilder boolQuery) {
		SearchRequestBuilder prepareSearch = mssEsClient.prepareSearch(tableNames.split(","));

		prepareSearch.setQuery(boolQuery);
		if (featchSize >= 0) {
			prepareSearch.setSize(featchSize);
		}

		AggregationBuilder<?> tb = makeAgg(field[0], size);

		AggregationBuilder<?> ftb = tb;
		AggregationBuilder<?> stb = null;
		for (int i = 1; i < field.length; i++) {
			stb = makeAgg(field[i], size);
			ftb.subAggregation(stb);
			ftb = stb;
		}

		if (ftb instanceof TermsBuilder) {
			((TermsBuilder) ftb).size(size);
		}

		return prepareSearch.setSize(0).addAggregation(tb).get();
	}

	private AggregationBuilder<?> makeAgg(String field, int size) {
		if (field.contains("|")) {
			String[] split = field.split("\\|");
			if ("date_histogram".equals(split[0])) {
				return AggregationBuilders.dateHistogram(split[1]).interval(new DateHistogramInterval(split[2]))
						.format(split[3]).field(split[1]);
			} else if ("nested".equals(split[0])) {

				AggregationBuilder<?> head = null;
				if (split.length == 3) {
					head = AggregationBuilders.terms(split[1]).field(split[1] + "." + split[2]).size(size);
				} else {
					head = AggregationBuilders.terms(split[2]).field(split[1] + "." + split[2]).size(size);
				}

				AggregationBuilder<?> agg = head;
				for (int i = 3; i < split.length; i++) {
					TermsBuilder subAggregation = AggregationBuilders.terms(split[i]).field(split[1] + "." + split[i])
							.size(size);
					agg.subAggregation(subAggregation);
					agg = subAggregation;
				}

				return AggregationBuilders.nested(split[1]).path(split[1]).subAggregation(head);
			} else if ("terms".equals(split[0])) {
				if ("count".equals(split[2])) {
					return AggregationBuilders.terms(split[1]).field(split[1]).order(Order.count(true));
				} else {
					return AggregationBuilders.terms(split[1]).field(split[1])
							.order(Order.term(Boolean.valueOf(split[2])));
				}

			}
		} else {
			return AggregationBuilders.terms(field).field(field);
		}
		return null;
	}

	/**
	 * 通过id查找数据
	 *
	 * @param tableName
	 *            数据表名称
	 * @param includes
	 *            包含哪些字段title,author
	 * @param excludes
	 *            不包含哪些字段abstract,pubyear
	 * @param id
	 *            数据id
	 * @return 数据的json格式
	 */
	public Restful findById(String tableName, String includes, String excludes, String id) {
		GetRequestBuilder rb = mssEsClient.prepareGet(tableName, tableName, id);

		String[] in = null, ex = null;
		if (StringUtil.isNotBlank(includes)) {
			in = includes.split(",");
		}
		if (StringUtil.isNotBlank(excludes)) {
			ex = excludes.split(",");
		}

		if (StringUtil.isNotBlank(includes) || StringUtil.isNotBlank(excludes)) {
			rb.setFetchSource(in, ex);
		}
		return Restful.instance(rb.get().getSource());
	}

	/**
	 * 具体执行搜索的类
	 *
	 * @param tableNames
	 * @param from
	 * @param size
	 * @param explain
	 * @param includes
	 * @param excludes
	 * @param distinct
	 * @param queryDsl
	 * @return
	 */
	private Object _search(String tableNames, int from, int size, String defaultOperator, boolean explain,
			String includes, String excludes, String distinct, String queryDsl, String filterDsl, String sort) {

		BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

		// 默认运算符
		QueryStringQueryBuilder.Operator op = StringUtils.equalsIgnoreCase(QueryStringQueryBuilder.Operator.AND.name(),
				defaultOperator) ? QueryStringQueryBuilder.Operator.AND : QueryStringQueryBuilder.Operator.OR;

		if (StringUtil.isNotBlank(filterDsl)) {
			boolQuery.filter(
					QueryBuilders.queryStringQuery(filterDsl).defaultOperator(op).lowercaseExpandedTerms(false));
		}

		if (StringUtil.isNotBlank(queryDsl)) {
			boolQuery.must(QueryBuilders.queryStringQuery(queryDsl).defaultOperator(op).lowercaseExpandedTerms(false));
		}

		return _search(tableNames, from, size, explain, includes, excludes, distinct, sort, boolQuery);
	}

	// @org.nlpcn.jcoder.run.annotation.Cache
	private Object _search(String tableNames, int from, int size, boolean explain, String includes, String excludes,
			String distinct, String sort, QueryBuilder boolQuery) {
		if (StringUtil.isNotBlank(distinct)) {
			JSONObject dtObj = JSONObject.parseObject(distinct);
			SearchRequestBuilder srb = mssEsClient.prepareSearch(tableNames.split(","))
					.setSearchType(SearchType.DFS_QUERY_THEN_FETCH).setQuery(boolQuery).setFrom(0)
					.setSize(dtObj.getIntValue("fetchCount")).setExplain(explain);
			sort(sort, srb);
			srb.setProfile(true);

			JSONArray fields = dtObj.getJSONArray("fields");

			String[] din = fields.stream().map(j -> ((JSONObject) j).getString("field")).distinct()
					.toArray(String[]::new);
			srb.setFetchSource(din, null);
			SearchResponse actionGet = srb.get();

			JSONObject result = JSONObject.parseObject(actionGet.toString());

			JSONArray jsonArray = result.getJSONObject("hits").getJSONArray("hits");

			jsonArray = mergeSame(fields, jsonArray, dtObj.getBooleanValue("mergeList"));

			result.getJSONObject("hits").put("fetchCount", jsonArray.size());

			List<Object> subList = jsonArray.subList(from, Math.min(from + size, jsonArray.size()));

			String[] in = null, ex = null;
			if (StringUtil.isNotBlank(includes)) {
				in = includes.split(",");
			}
			if (StringUtil.isNotBlank(excludes)) {
				ex = excludes.split(",");
			}

			for (Object j : subList) {
				JSONObject job = (JSONObject) j;
				GetRequestBuilder rb = mssEsClient.prepareGet(job.getString("_index"),
						job.getString("_type"), job.getString("_id"));

				if (StringUtil.isNotBlank(includes) || StringUtil.isNotBlank(excludes)) {
					rb.setFetchSource(in, ex);
				}

				job.put("_source", rb.get().getSource());
			}

			result.getJSONObject("hits").put("hits", subList);

			return result;

		} else {

			SearchRequestBuilder srb = mssEsClient.prepareSearch(tableNames.split(","))
					.setSearchType(SearchType.DFS_QUERY_THEN_FETCH).setQuery(boolQuery).setFrom(from).setSize(size)
					.setExplain(explain);

			sort(sort, srb);

			/**
			 * 设置排除字段
			 */
			if (StringUtil.isNotBlank(includes) || StringUtil.isNotBlank(excludes)) {
				String[] in = null, ex = null;
				if (StringUtil.isNotBlank(includes)) {
					in = includes.split(",");
				}

				if (StringUtil.isNotBlank(excludes)) {
					ex = excludes.split(",");
				}
				srb.setFetchSource(in, ex);
			}

			SearchResponse actionGet = srb.get();
			return actionGet;
		}
	}

	/**
	 * 排序
	 *
	 * @param sort
	 *            pubyear|asc,_score|desc
	 * @param srb
	 */
	private void sort(String sort, SearchRequestBuilder srb) {
		if (StringUtil.isNotBlank(sort)) {
			String[] split = sort.split(",");
			for (String str : split) {
				String[] s1 = str.split("\\|");
				SortOrder so = null;
				if (s1.length == 1) {
					so = SortOrder.DESC;
				} else {
					so = SortOrder.valueOf(s1[1].toUpperCase());
				}
				srb.addSort(SortBuilders.fieldSort(s1[0]).order(so).missing("_last"));
			}
		}
	}

	private JSONArray mergeSame(JSONArray fields, JSONArray jsonArray, boolean mergeList) {

		boolean allInfo = false;

		for (Object f : fields) {
			JSONObject job = (JSONObject) f;
			if ("field".equals(job.getString("type"))) {
				allInfo = true;
				break;
			}
		}

		for (Object f : fields) {
			JSONObject job = (JSONObject) f;
			String type = job.getString("type");
			if ("hash".equals(type)) {
				mergeSameByHash(job, jsonArray, mergeList, allInfo);
			} else if ("JaroWinkler".equals(type)) {
				// mergeSameByJaroWinklerDistance(job, jsonArray, mergeList, allInfo);
			}
		}

		return jsonArray;
	}

	/**
	 * 用hash的方式进行合并
	 *
	 * @param job
	 * @param jsonArray
	 * @param mergeList
	 * @param allInfo
	 * @Param 是否保留所有的信息
	 */
	private void mergeSameByHash(JSONObject job, JSONArray jsonArray, boolean mergeList, boolean allInfo) {
		String field = job.getString("field");

		Map<String, JSONObject> map = new HashMap<>();

		Iterator<Object> iterator = jsonArray.iterator();

		while (iterator.hasNext()) {
			JSONObject next = (JSONObject) iterator.next();
			JSONObject source = next.getJSONObject("_source");
			String v = source.getString(field);
			if (v == null) { // null不被合并.
				continue;
			}
			// hash方式忽略大小写，及多空格替换为1个
			v = v.toLowerCase().trim().replaceAll("\\s+", " ");
			JSONObject jsonObject = map.get(v);
			if (jsonObject != null) {
				if (mergeList) {
					List<Object> list = (List<Object>) jsonObject.get("mergeList");
					if (allInfo) {
						list.add(next);
					} else {
						list.add(next.getString("_id"));
					}
				}
				iterator.remove();
			} else {
				if (mergeList) {
					next.put("mergeList", new ArrayList<>(1));
				}
				map.put(v, next);
			}
		}
	}

	/**
	 * 用Jaro-Winkler Distance的方式进行合并
	 */
	// private void mergeSameByJaroWinklerDistance(JSONObject job, JSONArray
	// jsonArray, boolean mergeList, boolean allInfo) {
	// String field = job.getString("field"), value;
	// double threshold = job.getDoubleValue("threshold");
	//
	// List<JSONObject> items = new ArrayList<>(jsonArray.size());
	//
	// Iterator<Object> iterator = jsonArray.iterator();
	//
	// JSONObject next, jsonObject;
	// while (iterator.hasNext()) {
	// next = (JSONObject) iterator.next();
	// value =
	// Optional.ofNullable(next.getJSONObject("_source").getString(field)).orElse(StringUtils.EMPTY).replaceAll("\\s+",
	// " ").trim().toLowerCase();
	//
	// //
	// if (StringUtils.isEmpty(value))
	// continue;
	//
	// //
	// String fVal = value;
	// jsonObject = items.parallelStream().filter(
	// item -> StringUtils.getJaroWinklerDistance(fVal,
	// item.getJSONObject("_source").getString(field).replaceAll("\\s+", "
	// ").trim().toLowerCase()) > threshold)
	// .findFirst().orElse(null);
	// if (jsonObject != null) {
	// if (mergeList) {
	// List<Object> list = (List<Object>) jsonObject.get("mergeList");
	// if (allInfo) {
	// list.add(next);
	// } else {
	// list.add(next.getString("_id"));
	// }
	// }
	// iterator.remove();
	// } else {
	// if (mergeList) {
	// next.put("mergeList", new JSONArray(1));
	// }
	// items.add(next);
	// }
	// }
	// }

	/**
	 * 根据某个文档的id查询
	 *
	 * @param tableName
	 *            文档表名称
	 * @param id
	 *            id
	 * @param recommendTableNames
	 *            推荐表名称.如果为空则为tableName
	 * @param size
	 *            返回条数
	 * @param includes
	 *            包含字段
	 * @return
	 * @throws IOException
	 */
	public Object recommendById(@Param("table_name") String tableName, @Param("id") String id,
			@Param("recommend_table_names") String[] recommendTableNames, @Param(value = "size", df = "10") int size,
			@Param("includes") String includes) throws IOException {

		JSONObject doc = (JSONObject) JSONObject.toJSON(findById(tableName, includes, null, id).getObj());

		if (recommendTableNames == null || recommendTableNames.length == 0) {
			recommendTableNames = new String[] { tableName };
		}

		JSONObject parseObject = JSONObject.parseObject(JSONObject.toJSONString(recommend(recommendTableNames,
				doc.toJSONString(), size, StringUtil.isBlank(includes) ? new String[0] : includes.split(","))));

		List<Object> list = new ArrayList<>();

		JSONArray jsonArray = parseObject.getJSONObject("obj").getJSONObject("hits").getJSONArray("hits");

		for (Object object : jsonArray) { // 移除相同id
			if (!id.equals(((JSONObject) object).getString("_id"))) {
				list.add(object);
			}
		}
		parseObject.getJSONObject("obj").getJSONObject("hits").put("hits", list);

		return parseObject;

	}

}
